package com.atguigu.userprofile.app

import java.util.Properties

import com.atguigu.userprofile.bean.TagInfo
import com.atguigu.userprofile.dao.TagInfoDAO
import com.atguigu.userprofile.util.{ClickHouseUtil, MyPropertiesUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object TaskExportCkApp {

  /**
   * 将每天跑的标签任务的结果组合的宽表 hdfs://hadoop102:8020/user_profile/user_profile0111/user_tag_merge_20210608
   *
   * @param args spark-submit执行任务,除了提交的参数,后面跟的 任务id , 日期
   */
  def main(args: Array[String]): Unit = {

    /*
        TODO Hive中标签任务宽表迁移到ck中.
         1、连接ck并根据每天计算的标签任务的标签名称建立 ck的表.其中表的字段和Hive中定义的宽表一致.需要注意的是:两个库中的数据类型不同,且ck需要指定建表拓展
         2、Hive中的表数据不能通过insert override的方式直接插入到ck中,因为两种数据库是异构的,只能通过共性jdbc通过程序控制数据写入.
         3、读取Hive中宽表的数据,通过执行SparkSQL的方式,插入到ck中.
     */

    val sparkConf: SparkConf = new SparkConf().setAppName("task_export_clickhouse_app") //.setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()


    /** 获取执行当前main方法的参数,参数1-taskId,参数2-执行日期 */
    val taskId: String = args(0)
    val taskDate: String = args(1)
    /** 定义ck中的表名: user_tag_merge_20200609 */
    val tableName = s"user_tag_merge_${taskDate.replace("-", "")}"

    /** 获得字段的列表,用于组合在ck中建表语句中的字段名称 */
    val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoListOnTask()
    val tagColNames = tagInfoList.map(_.tagCode.toLowerCase + " String ").mkString(",")
    /** 如果当前ck中存在这张表, 则删除 */
    val dropTableSql = s" drop table if exists $tableName";

    /*
      create table user_tag_merge_20200614
      (
          uid UInt64,
          tag_natural_characteristic_gender    String,
          tag_natural_characteristic_agegroup  String,
          tag_order_habit_order_last_30d_count String
      ) engine = MergeTree
           primary key  uid
           order by  uid
     */
    val createTableSQL =
      s"""
         |    create table  $tableName  ( uid UInt64, ${tagColNames} )
         |     engine = MergeTree
         |     primary key  uid
         |     order by  uid
       """.stripMargin

    ClickHouseUtil.executeSql(dropTableSql)
    ClickHouseUtil.executeSql(createTableSQL)

    /*
      TODO 2、读取要插入的数据hive  表 $tableName
     */
    val properties: Properties = MyPropertiesUtil.load("config.properties")
    val userprofileDbName = properties.getProperty("user-profile.dbname")
    /** 通过Spark-SQL 读取Hive中合并的标签任务宽表 */
    val queryTaskTagMergeByHive: DataFrame = sparkSession.sql(s"select * from $userprofileDbName.$tableName")

    /*
      TODO 3、把数据写入ClickHouse,先去Hive中查询画像标签任务执行后组合的宽表,再通过jdbc的方式往CK中写入数据.
     */

    val CLICKHOUSE_URL = properties.getProperty("clickhouse.url")
    queryTaskTagMergeByHive.write.mode(SaveMode.Append)
      .option("batchsize", "100") //在写入ck时,需要按照批次插入,防止一次插入一条,造成临时表频繁进行合并操作.这一点在使用ck中比较重要.
      .option("isolationLevel", "NONE") // 关闭事务,ck是OLAP数据库,不支持事务,需要手动关闭.
      .option("numPartitions", "4") // 设置并发,因为ck在设计时,使用的是多个cpu进行多线程并行,因此这里每个cpu设置的并发时,不要超过10.
      .option("driver", "ru.yandex.clickhouse.ClickHouseDriver")
      .jdbc(CLICKHOUSE_URL, tableName, new Properties())

  }
}
